package cn.edu.sdju.fengguoyu.RegisterServer;

import cn.edu.sdju.fengguoyu.*;
import cn.edu.sdju.fengguoyu.annotation.GrpcService;
import cn.edu.sdju.fengguoyu.proto.*;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import org.checkerframework.checker.units.qual.A;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.Date;
import java.util.Map;

@GrpcService
public class RegisterService extends RegisterServiceGrpc.RegisterServiceImplBase {

    public  static int counter = 0;
    Logger logger = LoggerFactory.getLogger(RegisterService.class);
    @Autowired
    RegisterServers registerServers;
    @Autowired
    UserService userService;
    @Autowired
    ObserverConnectionsPool observerConnectionsPool;

    @Autowired
    WaitingMap waitingMap;

    @Override
    public void queryServiceDetail(QueryRequest request, StreamObserver<ServicesDetailList> responseObserver) {
        logger.info("Some query ALL Service detail");
        ServicesDetailList.Builder listBuilder = ServicesDetailList.newBuilder();
        for (Map.Entry<String, ArrayList<RegisterServerInfo>> listEntry:
             registerServers.getRawData().entrySet()) {
            for (RegisterServerInfo info:
                    listEntry.getValue()) {
                listBuilder.addServices(
                        ServicesDetail.newBuilder()
                                .setId(info.getId())
                                .setRank(info.getRank())
                                .setType("NODE")
                                .setNamed(listEntry.getKey())
                                .setStatus(ServicesDetail.Status.OK)
                                .setIpAddress(info.getService().getIpAddress())
                                .setPort(info.getService().getPort())
                                .build()
                );
            }
        }
        responseObserver.onNext(listBuilder.build());
        responseObserver.onCompleted();
    }

    @Override
    public void getService(QueryRequest request, StreamObserver<ServiceResponse> responseObserver) {
        logger.info("getService for {}", request.getServiceName());
        RegisterServerInfo info = registerServers.getBestServer(request.getServiceName());
        if (info == null) {
            logger.info("No more server");
            waitingMap.streamObserverHashMap.put(request.getServiceName(), responseObserver);
            return;
        } else {
            logger.info("response: {}:{}", info.getService().getIpAddress(),info.getService().getPort());
            responseObserver.onNext(ServiceResponse.newBuilder().setIpAddress(info.getService().getIpAddress())
                    .setPort(info.getService().getPort()).setId(info.getId()).build());
        }
        responseObserver.onCompleted();
    }

    @Override
    public void queryServices(QueryRequest request, StreamObserver<ServicesList> responseObserver) {
        logger.info("queryServices for {}", request.getServiceName());
        ArrayList<Service> services = new ArrayList<>();
        if (registerServers.getRawData().containsKey(request.getServiceName())) {
            for (RegisterServerInfo info: registerServers.getRawData().get(request.getServiceName())) {
                services.add(info.getService());
            }
        }
        responseObserver.onNext(ServicesList.newBuilder().addAllServices(services).build());
        responseObserver.onCompleted();
    }

    @Override
    public void updateRank(UpdateRankRequest request, StreamObserver<Response> responseObserver) {
        logger.info("updateRank: {} new rank: {}",request.getId(), request.getNewRank());
        registerServers.updateRank(request.getId(), request.getNewRank());
    }

    @Override
    public void registerService(RegisterInfo request, StreamObserver<Response> responseObserver) {
        logger.info("got server  connection");
        int id = registerServers.register(request);
        logger.info("set server id = {}", id);
        responseObserver.onNext(Response.newBuilder().setServerId(id).setStatus(Response.Status.OK).build());
        responseObserver.onCompleted();

    }

    @Override
    public void registerUser(UserInfo request, StreamObserver<Response> responseObserver) {
        logger.info("got user  connection");
        int userId = userService.registerUser(request);
        logger.info("set user id = {}", userId);
        responseObserver.onNext(Response.newBuilder().setStatus(Response.Status.OK).setUserId(userId).build());
        responseObserver.onCompleted();
    }

    @Override
    public void updateUserStatus(UpdateUserStatusRequest request, StreamObserver<Response> responseObserver) {
        userService.updateUserStatusById(request);
    }

    @Override
    public void reportServiceStatus(ServiceStatusInfo request, StreamObserver<Response> responseObserver) {
        responseObserver.onNext(Response.newBuilder().setStatus(Response.Status.OK).build());
        responseObserver.onCompleted();
        if (request.getStatus() == ServiceStatusInfo.Status.ERROR) {
            logger.info("用户报告服务异常：{}[{}]", request.getNamed(), request.getServerId());
            registerServers.removeService(request.getNamed(), request.getServerId());
            logger.info("移除服务器{}[{}]", request.getNamed(), request.getServerId());
            for (ManagedChannel channel:
                    observerConnectionsPool.observerChannels.values()) {
                ObserverMessageBroadcastGrpc.ObserverMessageBroadcastBlockingStub stub = ObserverMessageBroadcastGrpc.newBlockingStub(channel);
                stub.sendServiceError(ServiceErrorMessage.newBuilder().setId(request.getServerId()).build());
            }
        }

    }

    @Override
    public void heartbeat(Ping request, StreamObserver<Timestamp> responseObserver) {
        RegisterServerInfo info = registerServers.getServerById(request.getServerId());
        if (info == null)
            return;
        logger.info("server[{}]: PING", request.getServerId());
        long stamp = new Date().getTime();
        info.setTimestamp(stamp);
        responseObserver.onNext(Timestamp.newBuilder().setStamp(stamp).build());
        responseObserver.onCompleted();
    }

    @Override
    public void registerObserver(ObserverInfo request, StreamObserver<Response> responseObserver) {
        logger.info("reg {}:{}", request.getIpAddress(), request.getPort());
        ManagedChannel channel = ManagedChannelBuilder.forAddress(request.getIpAddress(), request.getPort()).usePlaintext().build();
        channel.getState(true);
        observerConnectionsPool.observerChannels.put(counter, channel);
        responseObserver.onNext(Response.newBuilder().setStatus(Response.Status.OK).setUserId(counter).build());
        responseObserver.onCompleted();
    }

    @Override
    public void unregisterObserver(UnregisterObserverRequest request, StreamObserver<Response> responseObserver) {
        observerConnectionsPool.observerChannels.remove(request.getId());
        responseObserver.onNext(Response.newBuilder().setStatus(Response.Status.OK).build());
        responseObserver.onCompleted();
    }

    @Override
    public void unregisterUser(UserInfo request, StreamObserver<Response> responseObserver) {
        responseObserver.onNext(Response.newBuilder().setStatus(Response.Status.OK).build());
        responseObserver.onCompleted();
        logger.info("Some user leave");
        userService.unRegisterUser(request);
        int serviceId = userService.userServiceMap.get(request.getId());
        userService.userServiceMap.remove(request.getId());
        int oldRank = registerServers.getServerById(serviceId).getRank();
        registerServers.updateRank(serviceId, oldRank-1);
    }

    @Override
    public void reportConnected(ConnectedMessage request, StreamObserver<Response> responseObserver) {
        responseObserver.onNext(Response.newBuilder().setStatus(Response.Status.OK).build());
        responseObserver.onCompleted();
        userService.userServiceMap.put(request.getUserId(), request.getServiceId());
        for (ManagedChannel channel:
        observerConnectionsPool.observerChannels.values()) {
            ObserverMessageBroadcastGrpc.ObserverMessageBroadcastBlockingStub stub = ObserverMessageBroadcastGrpc.newBlockingStub(channel);
            stub.sendNewConnectionMessage(NewConnectionMessage.newBuilder().setUserId(request.getUserId()).setServiceId(request.getServiceId()).build());
        }
        int oldRank = registerServers.getServerById(request.getServiceId()).getRank();
        registerServers.updateRank(request.getServiceId(), oldRank+1);
    }
}
